// Copyright [2018] Alibaba Cloud All rights reserved
#ifndef ENGINE_RACE_ENGINE_RACE_H_
#define ENGINE_RACE_ENGINE_RACE_H_

#include <string>
#include <include/engine.h>
#include <sys/stat.h>
#include <thread>
#include "constants.h"
#include "KeyValueLogManager.h"
#include "KeyValueLog.h"
#include "RDPCache.h"

namespace polar_race {

    static char *prepare(size_t size) {
        auto buffer = static_cast<char *>(malloc(size));
        posix_memalign((void **) &buffer, size, size);
        return buffer;
    }

    static thread_local std::unique_ptr<char> readBuffer(prepare(4096));


    class EngineRace : public Engine {
    public:
        static RetCode Open(const std::string &name, Engine **eptr) {
            struct stat st = {0};
            if (stat(name.data(), &st) == -1) {
                mkdir(name.data(), 0777);
            }

            auto *engine_race = new EngineRace(name);

            *eptr = engine_race;
            return kSucc;
        };

        explicit EngineRace(const std::string &dir) : rdpCache(CacheNumber, CacheThreadNumber, PerCacheSize),
                                                      start(now()) {
            fprintf(stderr, "%d files, %d shards, %d caches, %d cache segment\n", KeyManagerNum, KeyLogTotalNum,
                    CacheNumber, CacheSegmentCount);
            auto step = KeyManagerNum / 64;
            for (int i = 0; i < KeyManagerNum; i++) {
                keyValueLogManagers.emplace_back(new KeyValueLogManager(dir, i));
            }
            std::thread t[64];
            for (int i = 0; i < 64; ++i) {
                auto p = keyValueLogManagers.begin() + i * step;
                void *readBufferForKey = nullptr;
                posix_memalign(&readBufferForKey, SegmentSize, SegmentSize);
                t[i] = std::thread(initialKeyValueLog, p, step, readBufferForKey);
            }

            for (auto &i : t) {
                i.join();
            }

            fprintf(stderr, "open database time: %lims\n", (now() - start).count());
        }

        ~EngineRace() {
            for (auto &manager:keyValueLogManagers)
                delete manager;
            keyValueLogManagers.clear();
            fprintf(stderr, "database is closed. Total time: %lims\n", (now() - start).count());
        };

        KeyValueLog *selectKVLog(const PolarString &k, uint64_t *key) {
            stringToUint64(k.data(), key);
            auto index = *key >> KeyRightShiftNum;
            auto id = index / KeyLogPerManagerNum;
            auto valueLogId = index % KeyLogPerManagerNum;
            return keyValueLogManagers[id]->keyValueLogs[valueLogId];
        }

        RetCode Write(const PolarString &key,
                      const PolarString &value) {
            uint64_t key64 = 0;
            auto keyValueLog = selectKVLog(key, &key64);
            keyValueLog->putValue(key, value, key64);
            return kSucc;
        };

        RetCode Read(const PolarString &key,
                     std::string *value) {
            uint64_t key64 = 0;
            auto keyValueLog = selectKVLog(key, &key64);
            auto buffer = readBuffer.get();
            auto code = keyValueLog->readValue(key, buffer, key64);
            value->assign(buffer, 4096);
            return code;
        };

        /*
         * NOTICE: Implement 'Range' in quarter-final,
         *         you can skip it in preliminary.
         */
        RetCode Range(const PolarString &lower,
                      const PolarString &upper,
                      Visitor &visitor) {
            if (!lower.empty() || !upper.empty()) {
                return kNotSupported;
            }
            if (keyValueLogManagers[0]->keyValueLogs[0]->keyMap.size() > 20000) {
                if (!rdpCache.caching) {
                    rdpCache.startCaching(keyValueLogManagers, KeyManagerNum);
                }
                for (int id = 0; id < KeyManagerNum; id++) {
                    for (int i = 0; i < KeyLogPerManagerNum; i++) {
                        KeyValueLog *keyValueLog = keyValueLogManagers[id]->keyValueLogs[i];
                        KeyLogCache *cache = nullptr;
                        while (cache == nullptr) {
                            cache = rdpCache.getKeyLogCacheToRead(keyValueLog->id);
                            usleep(1);
                        }
                        cache->waitCacheReady();
                        keyValueLog->rangeWithCache(visitor, cache);
                        cache->releaseReference();
                    }
                }
            } else {
                for (int id = 0; id < KeyManagerNum; id++) {
                    for (int i = 0; i < KeyLogPerManagerNum; i++) {
                        KeyValueLog *keyValueLog = keyValueLogManagers[id]->keyValueLogs[i];
                        keyValueLog->normalRange(visitor, readBuffer.get());
                    }
                }
            }
            return kSucc;
        };

    private:
        milliseconds start{};
        RDPCache rdpCache;
        std::vector<KeyValueLogManager *> keyValueLogManagers;

        static void
        initialKeyValueLog(std::vector<KeyValueLogManager *>::iterator pKeyValueLog, int step, void *buffer) {
            for (int i = 0; i < step; i++) {
                (*(pKeyValueLog + i))->initialize(buffer);
            }
            free(buffer);
        }
    };

}  // namespace polar_race

#endif  // ENGINE_RACE_ENGINE_RACE_H_
